package com.snbc.util;

import com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.QualifierFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.List;

public class HBaseUtil {
	
	private static Logger logger = Logger.getLogger(HBaseUtil.class);
	private static Configuration config = HBaseConfiguration.create();
	private static HConnection connection = null;
	
	static {
		try {
			connection = HConnectionManager.createConnection(config);
		} catch (ZooKeeperConnectionException e) {
			logger.fatal("Create hbase connection failed, "+e.getMessage()+" :", e);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * @param tableName
	 * @brief 灏嗘暟鎹瓨鍏ュ埌HBase鐨勮〃涓�
	 * @param rowKey
	 * @param family
	 * @param qualifier
	 * @param value
	 * @throws IOException
	 */
	public static void put(String tableName, List<Put> puts){
		HTableInterface hTable = null;
		try {
			hTable = connection.getTable(tableName);
			hTable.put(puts);
		} catch(Exception e) {
			logger.fatal("Write hbase failed, "+e.getMessage()+" :", e);
		}finally {
			try {
				if (null != hTable) {
					hTable.close();
				}
			} catch (Exception e) {
				logger.fatal("Close hbase failed, "+e.getMessage()+" :", e);
			}
		}
	}
	public static void put(String tableName, Put puts){
		HTableInterface hTable = null;
		try {
			hTable = connection.getTable(tableName);
			hTable.put(puts);
		} catch(Exception e) {
			logger.fatal("Write hbase failed, "+e.getMessage()+" :", e);
		}finally {
			try {
				if (null != hTable) {
					hTable.close();
				}
			} catch (Exception e) {
				logger.fatal("Close hbase failed, "+e.getMessage()+" :", e);
			}
		}
	}



	
	private static void setGet(Get get, String family, String[] qualifiers, Long ts, int maxVersions, String[] excludeQualifiers) throws IOException, DeserializationException{
		if(maxVersions > 0){
			get.setMaxVersions(maxVersions);
		}
		if (null != family && null != qualifiers) {
			for(String qualifier :qualifiers){
				get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
			}
		}
		if (null != family && null == qualifiers) {
			get.addFamily(Bytes.toBytes(family));
		}
		if(null != ts){
			get.setTimeStamp(ts);
		}
		if(qualifiers==null && excludeQualifiers!=null && excludeQualifiers.length>0){
			Filter filter=null;
			for(String q:excludeQualifiers){
				filter = new QualifierFilter(CompareOp.NOT_EQUAL,new BinaryComparator(Bytes.toBytes(q)));
				get.setFilter(filter);
			}
		}
	}
	
	public static Result[] get(String tableName, List<String> rowKeys, String family) throws IOException {
		String[] excludeQualifiers={"bankNoteNoImage"};
		return get(tableName,rowKeys,family,null,null,1,excludeQualifiers);
	}
	/**
	 * 
	 * @param tableName  表名
	 * @param rowKeys    rowkey集合
	 * @param family    列族
	 * @param qualifiers  返回字段，若值为null，表示返回所有字段
	 * @param ts          指定返回记录的时间戳
	 * @param maxVersions  最大返回版本数
	 * @return
	 * @throws IOException
	 */
	public static Result[] get(String tableName, List<String> rowKeys, String family, String[] qualifiers, Long ts, int maxVersions, String[] excludeQualifiers) throws IOException {
		HTableInterface hTable = null;
		Result[] results = null;
		try {
			hTable = connection.getTable(tableName);
			List<Get> gets=Lists.newArrayList();
			Get get = null;
			for(String rowKey:rowKeys){
				get = new Get(Bytes.toBytes(rowKey));
				setGet(get,family,qualifiers,ts,maxVersions,excludeQualifiers);
				gets.add(get);
			}
			results = hTable.get(gets);
		} catch(Exception e) {
			logger.fatal("Read hbase table failed, "+e.getMessage()+" :", e);
		} finally {
			try {
				if (null != hTable) {
					hTable.close();
				}
			} catch (Exception e) {
				logger.fatal("Close hbase failed, "+e.getMessage()+" :", e);
			}
		}
		return results;
	}


	public static Result get(String tableName, String rowKey, String family) throws IOException {
		String[] excludeQualifiers={"bankNoteNoImage"};
		return get(tableName,rowKey,family,null,null,1,excludeQualifiers);
	}
	public static Result get(String tableName, String rowKey, String family, String[] qualifiers, Long ts, int maxVersions, String[] excludeQualifiers) throws IOException {
		HTableInterface hTable = null;
		Result result=null;
		try {
			hTable = connection.getTable(tableName);
			Get get = new Get(Bytes.toBytes(rowKey));
			setGet(get,family,qualifiers,ts,maxVersions,excludeQualifiers);
			result = hTable.get(get);
		} catch(Exception e) {
			logger.fatal("Read hbase table failed, "+e.getMessage()+" :", e);
		} finally {
			try {
				if (null != hTable) {
					hTable.close();
				}
			} catch (Exception e) {
				logger.fatal("Close hbase failed, "+e.getMessage()+" :", e);
			}
		}
		return result;
	}



	


	
	public static void truncateTable(String[] tableNames){
		HBaseAdmin hBaseAdmin = null;
		try {
			hBaseAdmin = new HBaseAdmin(config);
			for(String tableName:tableNames){
				if (hBaseAdmin.tableExists(tableName)) {  
	                hBaseAdmin.disableTable(tableName);  
	    			hBaseAdmin.truncateTable(TableName.valueOf(tableName), false);
//	    			hBaseAdmin.enableTable(tableName);
	            }
			}
		} catch (MasterNotRunningException e) {
			e.printStackTrace();
		} catch (ZooKeeperConnectionException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}finally{
			if(hBaseAdmin!=null){
				try {
					hBaseAdmin.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}

}
